package com.imu.purse.mq.conumer;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.Queues;
import com.imu.purse.modules.bitcoin.BitcoinApi;
import com.imu.purse.modules.trade.form.TransForm;
import com.imu.purse.modules.trade.service.TradeInfoService;
import com.imu.purse.mq.produce.TransProduce;
import com.rabbitmq.client.Channel;
import com.xiaoleilu.hutool.log.Log;
import com.xiaoleilu.hutool.log.LogFactory;
import org.apache.commons.lang.StringUtils;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.Iterator;
import java.util.Map;
import java.util.TimeZone;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

@Component
public class TransConsumer {
    private static final Log log = LogFactory.get(TransConsumer.class);

    @Autowired
    private BitcoinApi bitcoinApi;

    @Autowired
    private TransProduce transProduce;

    @Autowired
    private TradeInfoService tradeInfoService;


    //构建一个20个队列
    private static final ArrayBlockingQueue<String> transQueue = Queues.newArrayBlockingQueue(20);

    private static AtomicInteger transNum = new AtomicInteger(0);

    /* DIRECT模式.
     * @param message the message
     * @param channel the channel
     */
    @RabbitListener(queues = {"TRANS_OUT_QUEUE"})
    public void message(Message message, Channel channel) {
        try{
            //解析消息
            parseData(message);
        }catch(Exception e){//抛出异常，则该条消息需要重新存入数据库中
            log.error(e);
            //解析消息
            sendToProduce(message);
        }finally {
            try{
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            }catch (Exception e){
                //解析消息
                sendToProduce(message);
                log.error(e);
            }
        }
    }

    /**
     * 如果有问题重新丢入队列中
     * @param message
     */
    private void sendToProduce(Message message){
        if(message != null && message.getBody() != null){
            String data = new String(message.getBody());
            if(StringUtils.isNotEmpty(data)){
                //将消息转换为对象
                TransForm bean = JSON.parseObject(data, TransForm.class);
                if(bean == null){
                    return;
                }
                transProduce.sendData(bean);
            }
        }
    }

    /**
     * 分析数据
     * @param message
     */
    private void parseData(Message message){

        if(message != null && message.getBody() != null){
            String data = new String(message.getBody());
            if(StringUtils.isNotEmpty(data)){

                //将消息转换为对象
                TransForm bean = JSON.parseObject(data, TransForm.class);
                if(bean == null){
                    return;
                }
                if(transNum.intValue() < 20){
                    transNum.getAndIncrement();

                    String transId = null;
                    transId = bitcoinApi.sendfrom(bean.getPayPhone(), bean.getRecAddr(),
                            bean.getMoney(), bean.getFee(), bean.getCoreAddr());
                    try {
                        //调用核心钱包发送数据
                        //获取核心钱包
                        if (transId == null) {//调用失败
                            transProduce.sendData(bean);
                            sleepTime(2L);
                        }else if(StringUtils.contains(transId,"Insufficient funds")){//核心钱包出现假死的情况
                            transProduce.sendData(bean);
                            sleepTime(50L);
                        } else if(!StringUtils.startsWith(transId,"error")){//处理成功
                            transQueue.put(transId + "|" + bean.getCoreAddr());
                            bean.setTransId(transId);
                            tradeInfoService.transOutData(bean);
                            sleepTime(2L);
                        }else{//其他错误
                            transProduce.sendData(bean);
                            sleepTime(30L);
                        }
                    } catch (Exception e) {
                        log.error(e);
                        transProduce.sendData(bean);
                    }

                }else{
                    synchronized (transQueue){
                        if(transQueue.size() < 20){
                            transNum.set(transQueue.size());
                        }
                    }
                    transProduce.sendData(bean);
                }
            }
        }
    }

    /**
     * 休眠制定的时间
     * @param sleepTime
     */
    private void sleepTime(long sleepTime){
        try {
            TimeUnit.SECONDS.sleep(sleepTime);
        } catch (InterruptedException e) {
            log.error(e);
        }
    }
    /**
     * 处理已交易完的交易
     */
    private void completeTrans(){
        Map<String,Object> tMap = null;
        Iterator<String> iter = transQueue.iterator();
        String[] tmps = null;

        while (iter.hasNext()){
            tmps = StringUtils.split(iter.next(),"|");
            tMap =  bitcoinApi.gettransaction(tmps[0],tmps[1]);
            if(CollectionUtils.isEmpty(tMap)){
                continue;
            }
            if(tMap.get("error") != null || tMap.get("details") == null){
                continue;
            }
            iter.remove();
            transNum.getAndDecrement();
        }
    }

    @Scheduled(fixedRate = 100)
    public void clearCompleteTrans(){
        completeTrans();
    }
}
